package per.wxp.kafka.web;


import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import io.swagger.annotations.ApiImplicitParam;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;

import org.springframework.context.annotation.Bean;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.*;
import io.swagger.annotations.ApiOperation;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;
import per.wxp.kafka.entity.Message;
import per.wxp.kafka.service.KafkaService;

import javax.annotation.Resource;
import java.util.*;

@RestController
@RequestMapping(value="/kafka")
public class IndexController {

    protected final Logger logger = LoggerFactory.getLogger(this.getClass());


    @Resource
    private KafkaService kafkaService;

    @Autowired
    private Listener listener;


    private Gson gson = new GsonBuilder().create();

    @ApiOperation(value="发送信息", notes="根据KfakaVo对象发送信息")
    @ApiImplicitParam(name = "KfakaVo", value = "Kfakauser", required = true, dataType = "KfakaVo")
    @RequestMapping(value="/send", method=RequestMethod.POST)
    public ResponseEntity postUser(@RequestBody Message entity) {

        String[] arr={"ios","android","other"};
        Random random=new Random();
        Message m = new Message();
        m.setUid(UUID.randomUUID().toString().replace(" ",""));
        m.setClick_count(random.nextInt(20));
        m.setEvent_time(new Date().toString());
        m.setOs_type(arr[random.nextInt(2)]);
        try {
            logger.info("kafka的消息={}", m.toString());
            kafkaService.sendKafka("test",m);
            return new ResponseEntity( HttpStatus.OK);
        }catch (Exception e){
            logger.error("发送kafka失败", e);
            return  new ResponseEntity(HttpStatus.INTERNAL_SERVER_ERROR);
        }
    }

    private Map<String,Object> map= new HashMap<String,Object>();

    @ApiOperation(value="接收信息", notes="根据KfakaVo对象接受信息")
    @RequestMapping(value="/consume", method=RequestMethod.GET)
    public Map<String,Object> consume() {
        return map;
    }

    @KafkaListener(topics = {"test"})
    public void listen(ConsumerRecord<?, ?> record) {
        logger.info("kafka的key: " + record.key());
        logger.info("kafka的value: " + record.value().toString());
        Gson gson = new Gson();
        Message message  = gson.fromJson(record.value().toString(), Message.class);//对于javabean直接给出class实例
        map.put("key",message.getId());
        map.put("value",message.getDetails());
    }

}

